package cn.edu.flink.scala.tutorial.sql

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api.bridge.scala._

object Test {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource)

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    tEnv.createTemporaryView("click_table", clickStream,
      $("user"),
      $("url"),
      $("ts").proctime())

      tEnv.sqlQuery(
        s"""
           |select user, url, ts
           |from click_table
           |""".stripMargin).execute().print()

    env.execute("Test")
  }
}
